package com.hub.realtime.flink.submit.impl;

import com.hub.realtime.common.exception.UtilException;
import com.hub.realtime.common.model.FlinkRunRequest;
import com.hub.realtime.common.model.FlinkSubmitResponse;
import com.hub.realtime.common.utils.hadoop.HdfsUtil;
import com.hub.realtime.flink.submit.base.FlinkYarnSubmitBase;
import com.hub.realtime.flink.submit.model.FlinkSubmitCommon;
import com.hub.realtime.flink.submit.model.YarnRunEnv;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.client.deployment.ClusterClientFactory;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.yarn.YarnClusterDescriptor;
import org.apache.hadoop.yarn.api.records.ApplicationId;

import java.net.URI;
import java.util.Collections;
import java.util.List;

import static com.hub.realtime.common.constant.HdfsConstants.*;

@Slf4j
public class FlinkYarnApplicationSubmit extends FlinkYarnSubmitBase {
    @Override
    public FlinkSubmitResponse submit(FlinkRunRequest flinkRunRequest) {
        try {
            FlinkSubmitCommon flinkSubmitCommon = initalCommon(flinkRunRequest);
            URI uri = PackagedProgramUtils.resolveURI(flinkRunRequest.getJarFile());
            List<String> jobJars = Collections.singletonList(uri.toString());
            flinkSubmitCommon.setJobJars(jobJars);
            Configuration effectiveConfiguration = getEffectiveConfiguration(flinkSubmitCommon, flinkRunRequest);
            String checkpointHdfs = HdfsUtil.getDefaultFS(flinkRunRequest.getClusterInfo()).concat(APP_CHECKPOINTS);
            effectiveConfiguration.setString("state.checkpoints.dir", checkpointHdfs);
            String savepointHdfs = HdfsUtil.getDefaultFS(flinkRunRequest.getClusterInfo()).concat(APP_SAVEPOINTS);
            effectiveConfiguration.setString("state.savepoints.dir", savepointHdfs);
            String jobmanagerFs = HdfsUtil.getDefaultFS(flinkRunRequest.getClusterInfo()).concat(APP_FLINK_JOBS);
            effectiveConfiguration.setString("jobmanager.archive.fs.dir", jobmanagerFs);
            ApplicationConfiguration applicationConfiguration = ApplicationConfiguration.fromConfiguration(effectiveConfiguration);
            YarnRunEnv yarnRunEnv = createYarnClusterDescriptor(effectiveConfiguration, flinkRunRequest);
            YarnClusterDescriptor yarnClusterDescriptor = yarnRunEnv.getYarnClusterDescriptor();
            ClusterClientFactory<ApplicationId> clusterClientFactory = yarnRunEnv.getClusterClientFactory();
            ClusterSpecification clusterSpecification =
                    clusterClientFactory.getClusterSpecification(effectiveConfiguration);
            ClusterClientProvider<ApplicationId> applicationIdClusterClientProvider = yarnClusterDescriptor.deployApplicationCluster(clusterSpecification, applicationConfiguration);
            ClusterClient<ApplicationId> clusterClient = applicationIdClusterClientProvider.getClusterClient();
            FlinkSubmitResponse flinkSubmitResponse = makeRunResponse(flinkRunRequest.getClusterInfo(), clusterClient, yarnClusterDescriptor);
            yarnClusterDescriptor.close();
            return flinkSubmitResponse;
        } catch (Exception ex) {
            log.error("提交任务失败：" + ex.getMessage());
            ex.printStackTrace();
            String err = ex.getCause().getMessage();
            FlinkSubmitResponse flinkSubmitResponse = new FlinkSubmitResponse();
            flinkSubmitResponse.setMsg(err);
            flinkSubmitResponse.setIsSubmitError(true);
            if (StringUtils.isNotEmpty(err) && err.contains("yarn logs -applicationId ")) {
                flinkSubmitResponse.setAppId(StringUtils.substringAfter(err, "yarn logs -applicationId "));
            }
            return flinkSubmitResponse;
        }
    }
}
